package process;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author ChinaManor
 * #Description Table_Demo01
 * #Date: 22/6/2021 11:47
 */
public class Table_Demo01 {
    public static void main(String[] args) throws Exception {

    //1.准备环境 创建流环境 和 流表环境，并行度设置为1
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //2.Source 创建数据集
        DataStream<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(1L, "diaper", 4),
                new Order(3L, "rubber", 2)));

        DataStream<Order> orderB = env.fromCollection(Arrays.asList(
                new Order(2L, "pen", 3),
                new Order(2L, "rubber", 3),
                new Order(4L, "beer", 1)));

        //3.注册表 将数据流转换成表

        // 通过fromDataStream将数据流转换成表
        Table table1 = tenv.fromDataStream(orderA, $("user"), $("product"), $("amount"));
            // 将数据流转换成 创建临时视图
        tenv.createTemporaryView("OrderB",orderB,$("user"), $("product"), $("amount"));
        System.out.println(table1);
        //4.执行查询，查询order1的amount>2并union all 上 order2的amount<2的数据生成表
        Table resultQuery = tenv.sqlQuery("SELECT * FROM "+table1+" WHERE amount>2 UNION ALL SELECT * FROM OrderB WHERE amount< 2");
        //4.1 将结果表转换成toAppendStream数据流
        DataStream<Order> stream = tenv.toAppendStream(resultQuery, Order.class);
        stream.print();
        //5.打印结果
        env.execute();
    //6.执行环境
    // 创建实体类 user:Long product:String amount:int


    }
    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order {
        public Long user;
        public String product;
        public int amount;
    }
}
